set hive.exec.dynamic.partition = true;
set hive.exec.dynamic.partition.mode = 'nonstrict';
set hive.exec.max.dynamic.partitions.pernode = 200;
set hive.exec.max.dynamic.partitions = 200;
with t_arrival_first as(
    select
        waybill_no
         , scan_site_code
         , scan_site
         , scan_time
         , join_time
         , package_code
    from(
            select
                waybill_no
                 , scan_site_code
                 , scan_site
                 , scan_time
                 ,package_code
                 , row_number() over (partition by waybill_no order by scan_time asc) bill_rank --按照单号最早的一定是始发中心
        ,cast('{{ execution_date | cst_ds }}' as date) as join_time
            from jms_dwd.dwd_tab_barscan_centerarrival_base_dt
            where dt BETWEEN date_add('{{ execution_date | cst_ds }}',-6) AND '{{ execution_date | cst_ds }}'
              and network_type=2 --1.加盟商,2.转运中心,3.集散点,4.普通网点,5.代理区,6.总部,0.其他'
              and scan_site_code not like 'B%'
        ) tmp where bill_rank = 1

    union all

    select
        waybill_no
         , scan_site_code
         , scan_site
         , scan_time
         , join_time
         , package_code
    from(
            select
                waybill_no
                 , scan_site_code
                 , scan_site
                 , scan_time
                 , package_code
                 , row_number() over (partition by waybill_no order by scan_time asc) bill_rank --按照单号最早的一定是始发中心
        ,date_add('{{ execution_date | cst_ds }}',-4) as join_time
            from jms_dwd.dwd_tab_barscan_centerarrival_base_dt
            where dt BETWEEN date_add('{{ execution_date | cst_ds }}',-10) AND date_add('{{ execution_date | cst_ds }}',-4)
              and network_type=2 --1.加盟商,2.转运中心,3.集散点,4.普通网点,5.代理区,6.总部,0.其他'
        ) tmp where bill_rank = 1
),

     t_send as(
         select
             waybill_no
              , scan_site_code
              , scan_site
              , scan_time
              , package_code
              , dt
         from(
             select waybill_no
                  , scan_site_code
                  , scan_site
                  , scan_time
                  , package_code
                  , date(scan_time) as dt
            , row_number() over (partition by scan_site_code,waybill_no order by scan_time asc) bill_rank --中心最晚发件
from jms_dwd.dwd_tab_barscan_centersend_base_dt
WHERE --(dt = '{{ execution_date | cst_ds }}' or dt = date_add('{{ execution_date | cst_ds }}',-4))
    dt between date_add('{{ execution_date | cst_ds }}',-4) and '{{ execution_date | cst_ds }}'
  and  (date(scan_time) = '{{ execution_date | cst_ds }}' or date(scan_time) = date_add('{{ execution_date | cst_ds }}',-4))
--and date(scan_time) ='{{ execution_date | cst_ds }}' 
  and next_network_type =2
  and nvl(cast(mobile AS decimal(13, 3)), 0) <= 30
  and nvl(cast(remark AS decimal(13, 3)), 0) <= 30
  and nvl(cast(remark1 AS decimal(13, 3)), 0) <= 30
    ) t where bill_rank = 1
    ),
    send_all as(
select
    t_arrival_first.waybill_no
        ,t_arrival_first.scan_site_code
        ,t_arrival_first.scan_site
        ,t_arrival_first.scan_time
        ,t_arrival_first.package_code as arr_package_code
        ,t_send.package_code  as send_package_code
        ,t_send.dt
from t_arrival_first
    inner join t_send on t_arrival_first.scan_site_code = t_send.scan_site_code
    and t_arrival_first.waybill_no = t_send.waybill_no
    and t_arrival_first.join_time = t_send.dt
    and t_arrival_first.scan_time <= t_send.scan_time
    ),
    tab_send_scan as(  --始发中心发件
select
    cs.scan_time
        ,cs.waybill_no
        ,cs.send_package_code
        ,cs.arr_package_code
        ,cs.scan_site_code
        ,cs.scan_site
        ,os.customer_code
        ,os.customer_name
        ,os.terminal_dispatch_code
        ,os.real_pick_network_code
    -- ,os.real_pick_network_name
        ,nvl(wb.package_length               ,0.0) as package_length
        ,nvl(wb.package_wide                 ,0.0) as package_wide
        ,nvl(wb.package_high                 ,0.0) as package_high
        ,wb.goods_type_code
        ,wb.goods_type_name
        ,nvl(wb.package_inside_charge_weight ,0.0) as package_inside_charge_weight
        ,split(os.terminal_dispatch_code,',')[0] first_code
        ,bag.scan_time as bagging_time  --建包扫描时间
        ,cs.dt
from send_all cs
    left join
    (   --订单表取三段码和实际揽收网点
    select   --订单表取实际实际取件网点和三段码
    orders.waybill_id
        ,orders.customer_code
        ,orders.customer_name
        ,orders.terminal_dispatch_code
        ,orders.real_pick_network_code
    --,orders.real_pick_network_name
    from jms_dwd.dwd_yl_oms_oms_order_dt orders
    where dt between date_add('{{ execution_date | cst_ds }}',-10) and  '{{ execution_date | cst_ds }}'
    ) os on cs.waybill_no =os.waybill_id
    left join
    (   --运单表取内部计费重量
    select
    waybill_no
    ,package_length
    ,package_wide
    ,package_high
    ,goods_type_code
    ,goods_type_name
    ,package_inside_charge_weight
    from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
    where dt between date_add('{{ execution_date | cst_ds }}',-10) and  '{{ execution_date | cst_ds }}'
    ) wb on cs.waybill_no=wb.waybill_no
    left join
    (  --建包扫描
    SELECT
    waybill_no
    ,scan_site_code
    ,scan_site
    ,scan_time
    ,package_code
    FROM
    (
    select
    waybill_no
    ,scan_site_code
    ,scan_site
    ,scan_time
    ,package_code
    ,row_number() over(partition by waybill_no, scan_site_code,package_code order by scan_time) rk
    from jms_dwd.dwd_tab_barscan_bagging_base_dt
    where dt between date_add('{{ execution_date | cst_ds }}',-10) and '{{ execution_date | cst_ds }}'
    )t  where rk=1
    )bag on bag.waybill_no=cs.waybill_no and bag.scan_site_code=cs.scan_site_code and bag.package_code=cs.send_package_code
    )
    ,
all_detail as (
select s.*
,net2.agent_code,net2.agent_name  --发件中心代理区
,fc.center_code as end_center_code,fc.center_name as end_center_name  --末端中心
,net1.agent_code as taking_agent_code,net1.agent_name as taking_agent_name --揽收网点代理区
,net1.name as real_pick_network_name --实际取件网点名称
,count(1) over(partition by s.first_code,s.scan_site_code) as flow_cnt
from tab_send_scan  s
left join jms_dim.dim_network_whole_massage net1 on s.real_pick_network_code=net1.code
left join jms_dim.dim_network_whole_massage net2 on net2.code =s.scan_site_code
left join ( select code,center_code,center_name
from  jms_dim.dim_lmdm_sys_first_code
where is_enable=1 and is_delete =1
group by code,center_code,center_name) fc on fc.code= s.first_code
),

    all_detail_tmp as (
    select
         scan_time      --发件时间
        ,waybill_no     --运单号
        ,send_package_code as package_code  --发件扫描包号
        ,scan_site_code --发件网点
        ,scan_site
        ,terminal_dispatch_code --订单表三段码
        ,real_pick_network_code --订单表取件网点 揽收网点
        ,real_pick_network_name --订单表取件网点 揽收网点
        ,package_length  --长
        ,package_wide    --宽
        ,package_high    --高
        ,goods_type_code  --物品类型
        ,goods_type_name  --物品类型
        ,package_inside_charge_weight --内部计费重量
        ,first_code       --一段ma
        ,flow_cnt         --单流向数量
        ,agent_code       --发件网点代理区
        ,agent_name
        ,end_center_code  --末端机构
        ,end_center_name
        ,taking_agent_code   --揽收网点代理区
        ,taking_agent_name
        ,bagging_time        --建包扫描时间
        ,customer_code
        ,customer_name
    --是否应建包：1应建 0不应建
        ,case --内部计费重量＞1KG或长宽高最大边＞30CM的生鲜、液体、易碎件
            when package_inside_charge_weight>1 and goods_type_code in ('bm000007','bm000009','bm000008')  then 0
            when (package_length>30 or package_wide>30 or package_high>30) and goods_type_code in ('bm000007','bm000009','bm000008')  then 0
            when package_inside_charge_weight>3 and goods_type_code not in ('bm000007','bm000009','bm000008')  then 0
            when (package_length>30 or package_wide>30 or package_high>30) and goods_type_code not in ('bm000007','bm000009','bm000008')  then 0
            --剔除【揽收网点（订单表）所属代理区不等于本中心所属代理区】的单号
            when taking_agent_code != agent_code then 0
            --【按快件一段码（订单表三段码字段中的一段码）进行分组，单流向小件数<10票】的单号
            when arr_package_code=send_package_code then 0
            when flow_cnt<10 then 0
            when package_inside_charge_weight<=3 and (package_length<=30 and package_wide<=30 and package_high<=30)=1 then 1
            end as is_should_bagging
        ,dt
from all_detail
    )

insert overwrite table jms_dm.dm_start_center_violate_detail_dt
select
    scan_time      --发件时间
     ,waybill_no     --运单号
     ,package_code   --发件扫描包号
     ,scan_site_code  center_code--发件网点
     ,scan_site as center_name
     ,terminal_dispatch_code --订单表三段码
     ,real_pick_network_code --订单表取件网点 揽收网点
     ,real_pick_network_name --订单表取件网点 揽收网点
     ,package_length  --长
     ,package_wide    --宽
     ,package_high    --高
     ,goods_type_code  --物品类型
     ,goods_type_name  --物品类型
     ,package_inside_charge_weight --内部计费重量
     ,first_code       --一段ma
     ,flow_cnt         --单流向数量
     ,agent_code       --发件网点代理区
     ,agent_name
     ,end_center_code  --末端机构
     ,end_center_name
     ,taking_agent_code   --揽收网点代理区
     ,taking_agent_name
     ,bagging_time        --建包扫描时间
--是否应建包：1应建 0不应建
     , is_should_bagging
--违规类型0
     ,case when is_should_bagging=0 and package_inside_charge_weight>1.0 and goods_type_code in ('bm000007','bm000009','bm000008')  and  package_code is not null then '重量违规'
           when is_should_bagging=0 and package_inside_charge_weight>3.0 and goods_type_code not in ('bm000007','bm000009','bm000008')  and  package_code is not null then '重量违规'
           when is_should_bagging=0 and sort_array(array(package_length,package_wide,package_high))[2] >30.0 and  package_code is not null then '边长违规'
           when is_should_bagging=0 and flow_cnt<10 and  package_code is not null then '单流向数量违规'
           else NULL
    end as volate_type
     ,customer_code
     ,customer_name
     ,dt as  date_time
     ,case when is_should_bagging=0 and package_inside_charge_weight>1.0 and goods_type_code in ('bm000007','bm000009','bm000008') and  package_code is not null then 1
           when is_should_bagging=0 and package_inside_charge_weight>3.0 and goods_type_code not in ('bm000007','bm000009','bm000008')  and  package_code is not null then 1
           when is_should_bagging=0 and sort_array(array(package_length,package_wide,package_high))[2] >30.0 and  package_code is not null then 1
           when is_should_bagging=0 and flow_cnt<10 and  package_code is not null then 1
           else 0
    end as is_violate_bagging
     ,dt
     ,case when is_should_bagging=0 and package_inside_charge_weight>1.0 and goods_type_code in ('bm000007','bm000009','bm000008') and  package_code is not null then 1
           when is_should_bagging=0 and package_inside_charge_weight>3.0 and goods_type_code not in ('bm000007','bm000009','bm000008')  and  package_code is not null then 1
           when is_should_bagging=0 and sort_array(array(package_length,package_wide,package_high))[2] >30.0 and  package_code is not null then 1
           when is_should_bagging=0 and flow_cnt<10 and  package_code is not null then 1
           else 0
    end as violate  --是否违规建包 二级分区 和is_violate_bagging 一致
from
    all_detail_tmp
distribute by pmod(hash(rand()), 34);
